package com.didd.core.common.utils.jctools;


import com.didd.core.common.utils.ObjectUtil;
import com.didd.core.common.utils.concurrent.threadpool.NamedThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.jctools.queues.MpmcUnboundedXaddArrayQueue;

import java.util.Queue;

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.locks.LockSupport;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

/***
 * @author diddd
 * @Date 2020/12/17
 * @Note 移植  jctools 的 MPMC 队列 ， 提到JDK 传统的 LinkBlockQueue队列 无锁+去线程池
 */
@Slf4j
public class MpmcUnboundedQueueExecutor {
    private static final AtomicIntegerFieldUpdater<MpmcUnboundedQueueExecutor> THREAD_CNT_UPDATER =
            AtomicIntegerFieldUpdater.newUpdater(MpmcUnboundedQueueExecutor.class, "threadCnt");
    private static final int DEFAULT_THREAD_CNT = 0;
    private static final int DEFAULT_THREAD_MAX_CNT = 10;
    private volatile int threadCnt = DEFAULT_THREAD_CNT;
    private volatile int threadMaxCnt;
    private static final int POLLCNT_THRESHOLD = 512;
    private static final int DEFAULT_CHUNK_SIZE = 512;
    private static final int MAX_QUEUE_SIZE = 5 * 10 * 1024;
    private FutureListener<?> futureListener;
    private MpmcUnboundedXaddArrayQueue<Runnable> mpmcUnboundedXaddArrayQueue;

    private ThreadFactory threadFactory;

    private String namedThreadFactory;

    private MpmcUnboundedQueueExecutor(int chunkSize, String namedThreadFactory, int threadMaxCnt, FutureListener<?> futureListener) {
        ObjectUtil.checkNotNull(namedThreadFactory, "线程工厂名不能为空");
        //ObjectUtil.checkNotNull(futureListener, "队列回调处理不能为空");
        if (threadMaxCnt <= 0) {
            throw new IllegalArgumentException("threadMaxCnt 最大线程数不能小于0");
        }
        this.futureListener = futureListener;
        mpmcUnboundedXaddArrayQueue = new MpmcUnboundedXaddArrayQueue(chunkSize);
        this.namedThreadFactory = namedThreadFactory;
        threadFactory = new NamedThreadFactory(namedThreadFactory, true);
        this.threadMaxCnt = threadMaxCnt;
    }

    public interface FutureListener<T> {
        void operationComplete(T t, Throwable e) throws Exception;
    }

    public static MpmcUnboundedQueueExecutorBuilder mpmcUnboundedQueueExecutorBuilder() {

        return new MpmcUnboundedQueueExecutorBuilder();
    }


    public static class MpmcUnboundedQueueExecutorBuilder {
        private int initialCapacity = DEFAULT_CHUNK_SIZE;
        private String namedThreadFactory;
        private int threadMaxCnt = DEFAULT_THREAD_MAX_CNT;
        private FutureListener<?> futureListener;

        public MpmcUnboundedQueueExecutorBuilder setMinSize(int minSize) {
            this.initialCapacity = minSize;
            return this;
        }

        public MpmcUnboundedQueueExecutorBuilder setFutureListener(FutureListener futureListener) {
            this.futureListener = futureListener;
            return this;
        }

        public MpmcUnboundedQueueExecutorBuilder setThreadMaxCnt(int threadMaxCnt) {
            this.threadMaxCnt = threadMaxCnt;
            return this;
        }

        public MpmcUnboundedQueueExecutorBuilder setNamedThreadFactory(String namedThreadFactory) {
            this.namedThreadFactory = namedThreadFactory;
            return this;
        }

        public MpmcUnboundedQueueExecutor build() {
            return new MpmcUnboundedQueueExecutor(initialCapacity, namedThreadFactory, threadMaxCnt, futureListener);
        }
    }

    public void submit(Runnable runnable) {
        int threadCnt = THREAD_CNT_UPDATER.get(this);
        if (threadCnt >= 0 && threadCnt <= threadMaxCnt) {
            //多做一步，减少 increment 损耗和并发，在第一步进行get获取值操作判断
            if (THREAD_CNT_UPDATER.incrementAndGet(this) <= threadMaxCnt) {
                addWorker();
            }
        }
        //控制一个队列 大小的估算值,在5万范围内，可以接受，主要防止 溢出
        if (mpmcUnboundedXaddArrayQueue.size() <= MAX_QUEUE_SIZE) {
            mpmcUnboundedXaddArrayQueue.offer(runnable);
        } else {
            log.error("mpmcUnboundedXaddArrayQueue 队列超过默认阀值5万 ，线程工厂{}", namedThreadFactory);
        }

    }

    private void addWorker() {
        //消费者线程
        threadFactory.newThread(new Runnable() {
            @Override
            public void run() {
                executeTask(mpmcUnboundedXaddArrayQueue);
            }
        }).start();

    }

    private void executeTask(Queue<Runnable> taskQueue) {
        int pollCnt = 0;
        for (; ; ) {
            try {
                // 防止空轮询，导致cpu 飙升
                if (!hasTasks()) {
                    pollCnt++;
                    // 防止空轮询，导致cpu 飙升 ,空轮询超过一个阀值，park 1秒的当前线程。
                    if (pollCnt >= POLLCNT_THRESHOLD * 1.25) {
                        pollCnt = 0;
                        //防止空轮询，导致cpu 飙升
                        LockSupport.parkNanos(MILLISECONDS.toNanos(1000));
                    }
                    continue;
                }
                Runnable task = taskQueue.poll();
                try {
                    if (task != null) {
                        safeExecute(task);
                    }
                } catch (Throwable t) {
                    log.error("taskQueue error", t);
                }
            } catch (Throwable e) {
                log.error("executeTask error", e);
            }
        }
    }

    protected boolean hasTasks() {
        return !mpmcUnboundedXaddArrayQueue.isEmpty();
    }

    /**
     * Try to execute the given {@link Runnable} and just log if it throws a {@link Throwable}.
     */
    protected static void safeExecute(Runnable task) {
        try {
            task.run();
        } catch (Throwable t) {
            log.warn("A task raised an exception. Task: {}", task, t);
        }
    }
}
